package io.plus.engine.pulse.task;

import io.plus.engine.pulse.*;
import io.plus.engine.pulse.task.queue.Queue;
import io.plus.engine.pulse.task.queue.QueueImpl;
import io.plus.engine.utils.lock_capable_task.AbstractQueueTask;
import lombok.extern.slf4j.Slf4j;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public class SendTask extends AbstractQueueTask<PacketMetadata> {

    public static final int MAX_DATA_LENGTH = 132476;// 85732; // 65535 - 40;//51200  //179743  //197301  //197301  //85732
    private static final int MAX_CAPACITY_MIN = MAX_DATA_LENGTH - 1024;    //减掉1024是尽量防止溢出的一小部分还分成一个tcp包发出
    private static final int MAX_CAPACITY_MAX = MAX_CAPACITY_MIN * 10;
    public boolean canSend = true;
    private AsynchronousSocketChannelContext asynchronousSocketChannelContext = null;
    private Configuration configuration = null;
    private Handler handler = null;

    private Queue<PacketMetadata> queue = null;

    public SendTask(AsynchronousSocketChannelContext asynchronousSocketChannelContext, Executor executor) {
        super(executor);
        this.asynchronousSocketChannelContext = asynchronousSocketChannelContext;
        this.configuration = asynchronousSocketChannelContext.clientConfiguration;
        this.handler = configuration.getHandler();

        getQueue();
    }

    @Override
    public Queue<PacketMetadata> getQueue() {
        if (queue == null) {
            synchronized (this) {
                if (queue == null) {
                    queue = new QueueImpl<PacketMetadata>();
                }
            }
        }
        return queue;
    }

    @Override
    public boolean add(PacketMetadata packetMetadata) {
        if (this.isCanceled()) {
            log.info("{}, 任务已经取消，{}添加到发送队列失败", asynchronousSocketChannelContext, packetMetadata.toString());
            return false;
        }
        return getQueue().add(packetMetadata);
    }

    /**
     * 清空消息队列
     */
    @Override
    public void clearQueue() {
        PacketMetadata packetMetadata = null;
        while ((packetMetadata = queue.poll()) != null) {
            try {
                asynchronousSocketChannelContext.processAfterSent(packetMetadata, false);
            } catch (Throwable e) {
                log.error(e.toString(), e);
            }
        }
    }


    @Override
    public void runTask() {
        if (queue.isEmpty()) {
            return;
        }

        int queueSize = queue.size();
        if (queueSize == 1) {
            PacketMetadata packetMetadata = queue.poll();
            if (packetMetadata != null) {
                sendPacket(packetMetadata);
            }
            return;
        }
        int listInitialCapacity = Math.min(queueSize, canSend ? 300 : 1000);
        PacketMetadata packetMetadata = null;
        List<PacketMetadata> packetMetadataList = new ArrayList<>(listInitialCapacity);
        List<ByteBuffer> byteBufferList = new ArrayList<>(listInitialCapacity);
        int allByteBufferCapacity = 0;
        boolean sslChanged = false;
        while ((packetMetadata = queue.poll()) != null) {
            ByteBuffer byteBuffer = getByteBuffer(packetMetadata);
            packetMetadataList.add(packetMetadata);
            byteBufferList.add(byteBuffer);
            allByteBufferCapacity += byteBuffer.limit();
            if ((canSend && allByteBufferCapacity >= MAX_CAPACITY_MIN) || (allByteBufferCapacity >= MAX_CAPACITY_MAX) || sslChanged) {
                break;
            }
        }
        if (allByteBufferCapacity == 0) {
            return;
        }
        ByteBuffer allByteBuffer = ByteBuffer.allocate(allByteBufferCapacity);
        for (ByteBuffer byteBuffer : byteBufferList) {
            allByteBuffer.put(byteBuffer);
        }
        allByteBuffer.flip();
        this.sendByteBuffer(allByteBuffer, packetMetadataList);
    }

    public boolean sendPacket(PacketMetadata packetMetadata) {
        ByteBuffer byteBuffer = getByteBuffer(packetMetadata);
        sendByteBuffer(byteBuffer, packetMetadata);
        return true;
    }

    private ByteBuffer getByteBuffer(PacketMetadata packetMetadata) {
        try {
            ByteBuffer byteBuffer = handler.encode(configuration, asynchronousSocketChannelContext, packetMetadata);

            if (!byteBuffer.hasRemaining()) {
                byteBuffer.flip();
            }
            return byteBuffer;
        } catch (Exception e) {
            log.error(packetMetadata.toString(), e);
            throw new RuntimeException(e);
        }
    }

    public void sendByteBuffer(ByteBuffer byteBuffer, Object packets) {
        if (byteBuffer == null) {
            log.error("{},byteBuffer is null", asynchronousSocketChannelContext);
            return;
        }
        if (!IoUtils.check(asynchronousSocketChannelContext)) {
            return;
        }
        ReentrantLock lock = asynchronousSocketChannelContext.writeCompletionHandler.lock;
        lock.lock();
        try {
            canSend = false;
            WriteCompletionVo writeCompletionVo = new WriteCompletionVo(byteBuffer, packets);
            asynchronousSocketChannelContext.asynchronousSocketChannel.write(byteBuffer, writeCompletionVo, asynchronousSocketChannelContext.writeCompletionHandler);
            asynchronousSocketChannelContext.writeCompletionHandler.condition.await();
        } catch (InterruptedException e) {
            log.error(e.toString(), e);
        } finally {
            lock.unlock();
        }
    }


}
